package org.senRPC.client.consumer;
import io.netty.channel.Channel;
import org.senRPC.client.config.ConsumerConfig;
import org.senRPC.common.protocal.senRPCProtocol;
import org.senRPC.common.utils.ChannelGroup;
import org.senRPC.common.utils.NettyChannelGroup;
import org.senRPC.common.utils.UnresolvedAddress;
import org.senRPC.remoting.config.NettyClientConfig;
import org.senRPC.remoting.netty.NettyRemotingClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * #####消费端########
 * @description 默认的消费类
 */
public abstract class DefaultConsumer extends AbstractDefaultConsumer {

	private static final Logger logger = LoggerFactory.getLogger(DefaultConsumer.class);
	//两个netty客户端,一个处理register一个处理provider
	private NettyClientConfig registryClientConfig;
	private NettyClientConfig providerClientConfig;

	private ConsumerConfig consumerConfig;
	//netty的客户端,一个用于拉取服务一个用于消费服务
	protected NettyRemotingClient registryNettyRemotingClient;
	protected NettyRemotingClient providerNettyRemotingClient;
	private ConsumerManager consumerManager;
	private SubscribeManager subscribeManager;
	//注册中心的信道只有一个
	private Channel registyChannel;

	public DefaultConsumer(NettyClientConfig registryClientConfig, NettyClientConfig providerClientConfig, ConsumerConfig consumerConfig) {
		this.registryClientConfig = registryClientConfig;
		this.providerClientConfig = providerClientConfig;
		this.consumerConfig = consumerConfig;
		consumerManager = new ConsumerManager(this);
		initialize();
	}

	private void initialize() {
		//因为服务消费端可以直连provider，所以当传递过来的与注册中心连接的配置文件为空的时候，可以不初始化registryNettyRemotingClient,直接与provder连接就行
		//与注册中心只需要一个NettyRemotingClient就行
		if(null != this.registryClientConfig){
			this.registryNettyRemotingClient = new NettyRemotingClient(this.registryClientConfig);
			// 注册处理器
			registerProcessor();
		}
		// 需要与每个生产者之间建立NettyRemotingClient
		this.providerNettyRemotingClient = new NettyRemotingClient(this.providerClientConfig);

	}

	private void registerProcessor() {
		//把处理器和实际干活的线程
		this.registryNettyRemotingClient.registerProcessor(senRPCProtocol.SUBCRIBE_RESULT, new DefaultConsumerRegistryProcessor(this), null);
		this.registryNettyRemotingClient.registerProcessor(senRPCProtocol.SUBCRIBE_SERVICE_CANCEL, new DefaultConsumerRegistryProcessor(this), null);
		this.registryNettyRemotingClient.registerProcessor(senRPCProtocol.CHANGE_LOADBALANCE, new DefaultConsumerRegistryProcessor(this), null);
	}
	//这段代码主要的核心意思就是获取的信息之后，建立连接，因为连接是异步的，所以加了通知机制，建立成功之后，将健康的channel放入到group中去，然后就维护了所有的channel
	@Override
	public SubscribeManager subscribeService(final String service) {

	    //SubscribeManager是内部接口
		// 每次调用都是新建了一个subscribeManager
		this.subscribeManager = new SubscribeManagerImpl(service,consumerManager,this);
		return subscribeManager;
	}

	@Override
	public boolean addChannelGroup(String serviceName, ChannelGroup group) {
		return DefaultConsumer.super.addIfAbsent(serviceName, group);
	}

	@Override
	public boolean removeChannelGroup(String serviceName, ChannelGroup group) {
		return DefaultConsumer.super.removedIfAbsent(serviceName, group);
	}

	@Override
	public Channel directGetProviderByChannel(UnresolvedAddress address) throws InterruptedException {
		return this.providerNettyRemotingClient.getAndCreateChannel(address.getHost()+":"+address.getPort());
	}

	@Override
	public void start() {
		logger.info("######### consumer start.... #########");
		// 如果连接注册中心的client初始化成功的情况下，且连接注册中心的地址不为空的时候去尝试连接注册中心
		if (null != this.registryClientConfig && null != this.registryNettyRemotingClient) {
			this.registryNettyRemotingClient.start();
			// 获取到与注册中心集群的一个健康的的Netty 长连接的channel
			getOrUpdateHealthyChannel();
		}
		this.providerNettyRemotingClient.setreconnect(false);
		this.providerNettyRemotingClient.start();
	}

	//
	@Override
	public void getOrUpdateHealthyChannel() {

		//获取到注册中心的地址
		String addresses = this.registryClientConfig.getDefaultAddress();

		if (registyChannel != null && registyChannel.isActive() && registyChannel.isWritable())
			return;

		if (addresses == null || "".equals(addresses)) {
			logger.error("registry address is empty");
			return;
		}

		//与注册中心连接的时候重试次数
		int retryConnectionTimes = this.consumerConfig.getRetryConnectionRegistryTimes();
		//连接给每次注册中心的时候最大的超时时间
		long maxTimeout = this.consumerConfig.getMaxRetryConnectionRegsitryTime();

		String[] adds = addresses.split(",");

		for (int i = 0; i < adds.length; i++) {

			if (registyChannel != null && registyChannel.isActive() && registyChannel.isWritable())
				return;

			String currentAddress = adds[i];
			//开始计时
			final long beginTimestamp = System.currentTimeMillis();
			long endTimestamp = beginTimestamp;

			int times = 0;

			//当重试次数小于最大次数且每个实例重试的时间小于最大的时间的时候，不断重试
			for (; times < retryConnectionTimes && (endTimestamp - beginTimestamp) < maxTimeout; times++) {
				try {
					Channel channel = registryNettyRemotingClient.createChannel(currentAddress);
					if (channel != null && channel.isActive() && channel.isWritable()) {
						this.registyChannel = channel;
						break;
					} else {
						continue;
					}
				} catch (InterruptedException e) {
					logger.warn("connection registry center [{}] fail", currentAddress);
					endTimestamp = System.currentTimeMillis();
					continue;
				}
			}
		}
	}

	public NettyRemotingClient getRegistryNettyRemotingClient() {
		return registryNettyRemotingClient;
	}

	public void setRegistryNettyRemotingClient(NettyRemotingClient registryNettyRemotingClient) {
		this.registryNettyRemotingClient = registryNettyRemotingClient;
	}

	public Channel getRegistyChannel() {
		return registyChannel;
	}

	public void setRegistyChannel(Channel registyChannel) {
		this.registyChannel = registyChannel;
	}

	public ConsumerConfig getConsumerConfig() {
		return consumerConfig;
	}

	public void setConsumerConfig(ConsumerConfig consumerConfig) {
		this.consumerConfig = consumerConfig;
	}

	public NettyClientConfig getRegistryClientConfig() {
		return registryClientConfig;
	}

	public void setRegistryClientConfig(NettyClientConfig registryClientConfig) {
		this.registryClientConfig = registryClientConfig;
	}

	public ConsumerManager getConsumerManager() {
		return consumerManager;
	}

	public void setConsumerManager(ConsumerManager consumerManager) {
		this.consumerManager = consumerManager;
	}

	public NettyRemotingClient getProviderNettyRemotingClient() {
		return providerNettyRemotingClient;
	}

	public void setProviderNettyRemotingClient(NettyRemotingClient providerNettyRemotingClient) {
		this.providerNettyRemotingClient = providerNettyRemotingClient;
	}




	public ChannelGroup group(UnresolvedAddress address) {
		// 保证每个channel的地址是不同的,相同的channel返回相同的channel
		ChannelGroup group = addressGroups.get(address);
		if (group == null) {
			ChannelGroup newGroup = newChannelGroup(address);
			group = addressGroups.putIfAbsent(address, newGroup);
			if (group == null) {
				group = newGroup;
			}
		}
		return group;
	}

	private ChannelGroup newChannelGroup(UnresolvedAddress address) {
		return new NettyChannelGroup(address);
	}


}
